package com.leyantech.ecp.agenx.common.misc;

import com.leyantech.ecp.agenx.common.utils.CacheUtils;

import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigService;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.google.common.util.concurrent.RateLimiter;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.ext.web.RoutingContext;
import io.vertx.ext.web.handler.PlatformHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;

import java.util.Objects;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-10-31.
 */
@Slf4j
public class HttpLocalRateLimiter implements PlatformHandler {

  static final Config apolloConfig = ConfigService.getConfig("rateLimiter");

  static final Supplier<LoadingCache<Pair<String, String>, RateLimiter>> cache = CacheUtils
      .buildCache(
          "rateLimiter",
          pathAndKey -> {
            final String path = pathAndKey.getLeft();
            final String key = pathAndKey.getRight();
            Double rate = apolloConfig.getDoubleProperty(key, null);
            if (rate == null) {
              rate = apolloConfig.getDoubleProperty(path, 3.0d);
              log.info("load rate by path {} {}", path, rate);
            } else {
              log.info("load rate by key {} {}", key, rate);
            }
            return RateLimiter.create(rate);
          }
      );

  private final Function<RoutingContext, String> rateKeyFunc;

  public HttpLocalRateLimiter(Function<RoutingContext, String> rateKeyFunc) {
    this.rateKeyFunc = Objects.requireNonNull(rateKeyFunc, "rateKeyFunc is null.");
  }


  @Override
  public void handle(RoutingContext context) {
    final String path = StringUtils.defaultIfBlank(context.request().path(), "root");
    final String key = rateKeyFunc.apply(context);
    final RateLimiter rateLimiter = cache.get().get(Pair.of(path, key));
    if (rateLimiter.tryAcquire()) {
      context.next();
    } else {
      log.warn("path {} key {} refused by rate limit.", path, key);
      context.response()
          .setStatusCode(HttpResponseStatus.SERVICE_UNAVAILABLE.code())
          .send("refused by rate limiter");
    }
  }

}
